package test.threads.queue.stressed;

import java.util.concurrent.Executor;

import com.google.common.base.Throwables;
import com.lmax.disruptor.EventFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.common.*;
import test.threads.queue.common.FSM.State;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;

/**
 * @author cheremin
 * @since 01.03.12,  13:32
 */
public abstract class AbstractTask<E> implements ITask<E> {
    private static final Log log = LogFactory.getLog( AbstractTask.class );

    /**
     * Время ожидания завершения.
     * 5 секунд должно быть достаточно для всех!
     */
    private static final long MAX_WAIT_TIME = 5000L;

    private final int bufferSize;
    private final int eventsInBatch;

    protected AbstractTask() {
        this( TaskBenchmark.QUEUE_CAPACITY, TaskBenchmark.EVENTS_IN_BATCH );
    }

    protected AbstractTask( final int bufferSize ) {
        this( bufferSize, TaskBenchmark.EVENTS_IN_BATCH );
    }

    protected AbstractTask( final Config config ) {
        this(
                config.getAsInt( "task.buffer-size", TaskBenchmark.QUEUE_CAPACITY ),
                config.getAsInt( "task.events-in-batch", TaskBenchmark.EVENTS_IN_BATCH )
        );
    }

    protected AbstractTask( final int bufferSize,
                            final int eventsInBatch ) {
        checkArgument(
                eventsInBatch < bufferSize,
                "eventsInBatch(%s) must be < bufferSize(%s)",
                eventsInBatch, bufferSize
        );
        this.bufferSize = bufferSize;
        this.eventsInBatch = eventsInBatch;
    }

    private boolean initialized = false;
    private EventFactory<E> eventFactory = null;
    private BaseEnqueuer eventEnqueuer = null;
    private Executor threadPool = null;

    private boolean executing = false;

    @Override
    public synchronized final void initialize( final Executor threadPool,
                                               final IUsefulWorkEmulator<E>[] emulators,
                                               final EventFactory<E> factory ) {
        checkArgument( threadPool != null, "threadPool can't be null" );
        checkArgument( emulators != null, "emulators can't be null" );
        checkArgument( emulators.length > 0, "emulators.length(%s) can't be 0", emulators.length );

        checkState( !initialized, "Can't initialize more then once" );

        this.eventFactory = factory;
        this.threadPool = threadPool;
        try {
            eventEnqueuer = _initialize( threadPool, emulators, factory );
            checkState( eventEnqueuer != null, "eventEnqueuer can't be null" );
            threadPool.execute( eventEnqueuer );
        } catch ( Exception e ) {
            throw Throwables.propagate( e );
        }
        initialized = true;
    }

    public int bufferSize() {
        return bufferSize;
    }

    public int eventsInBatch() {
        return eventsInBatch;
    }

    public synchronized boolean isInitialized() {
        return initialized;
    }

    public synchronized EventFactory<E> eventFactory() {
        checkState( initialized, "Task is not initialized" );
        return eventFactory;
    }

    protected abstract BaseEnqueuer _initialize( final Executor threadPool,
                                                 final IUsefulWorkEmulator<E>[] emulators,
                                                 final EventFactory<E> factory ) throws Exception;

    @Override
    public final BenchmarkResult[] execute( final long millisToBogus,
                                            final long millisToMeasure,
                                            final int measurements ) {
        checkState( initialized, "Can't execute uninitialized task" );
        checkState( !executing, "Previous execution not finished yet" );
        checkState( eventEnqueuer != null, "eventEnqueuer can't be null" );
        checkArgument( measurements >= 1, "measurements(%s) can't be <1", measurements );
        executing = true;
        try {
            return executeImpl( millisToBogus, millisToMeasure, measurements );
        } catch ( Exception e ) {
            throw Throwables.propagate( e );
        } finally {
            executing = false;
        }
    }

    @Override
    public void terminate() {
        checkState( initialized, "Can't terminate uninitialized task" );
        checkState( eventEnqueuer != null, "eventEnqueuer can't be null" );
        eventEnqueuer.terminate();
    }

    protected BenchmarkResult[] executeImpl( final long millisToBogus,
                                             final long millisToMeasure,
                                             final int measurements ) throws Exception {
        eventEnqueuer.waitForInitialized();

        eventEnqueuer.start();
        eventEnqueuer.waitForStarting();

        //dequeue some packets before start timer to warm/fill up pipeline (not JIT!)
        final BenchmarkResult bogusResult = dequeueWhileTimeIsNotEllapsed( millisToBogus );

        final long msToOneMeasure = millisToMeasure / measurements;
        checkState(
                msToOneMeasure > 1,
                "msToMeasure(%s) = millisToMeasure(%s)/measurements(%s) must be > 1",
                msToOneMeasure, millisToMeasure, measurements
        );
        final BenchmarkResult[] results = new BenchmarkResult[measurements];
        long packetsCompleted = 0;
        for ( int i = 0; i < results.length; i++ ) {
            results[i] = dequeueWhileTimeIsNotEllapsed( msToOneMeasure );
            packetsCompleted += results[i].packetsCompleted;
        }

        long eventsDequeued = bogusResult.packetsCompleted + packetsCompleted;

        eventEnqueuer.pause();
        //wait for packets exhausted
        Thread.yield();
        //fixme: здесь периодически возникает ситуация, когда мы подвисаем. Понять причины точно не удается
        //но чтобы не подвешивать эксперименты я здесь просто ввел ограничение времени
        final long startedAt = System.currentTimeMillis();
        while ( !eventEnqueuer.is( State.WAITING ) ) {
            while ( eventEnqueuer.eventsEnqueued() > eventsDequeued ) {
//            log.info( "Waiting for " + eventEnqueuer.eventsEnqueued() + " > " + eventsDequeued );
                dequeueBatchEvents( eventsInBatch );
                eventsDequeued += eventsInBatch;
            }
            Thread.yield();
            final long current = System.currentTimeMillis();
            if ( current - startedAt > MAX_WAIT_TIME ) {
                //идите в жопу, я -- заканчиваю
                break;
            }
        }
        return results;
    }

    private BenchmarkResult dequeueWhileTimeIsNotEllapsed( final long timeLimit ) throws Exception {
        long eventsDequeued = 0;
        final long startedAt = System.currentTimeMillis();
        final long startedAtNs = System.nanoTime();
        while ( true ) {
            dequeueBatchEvents( eventsInBatch );
            eventsDequeued += eventsInBatch;

            final long finishedAtMs = System.currentTimeMillis();
            final long ellapsedMs = finishedAtMs - startedAt;
            if ( ellapsedMs >= timeLimit ) {
                final long finishedAtNs = System.nanoTime();
                final long ellapsedNs = finishedAtNs - startedAtNs;

                final BenchmarkResult result = new BenchmarkResult(
                        ellapsedMs,
                        ellapsedNs,
                        eventsDequeued
                );
                return result;
            }
        }
    }

    protected abstract long dequeueBatchEvents( final int eventsInBatch ) throws Exception;

    @Override
    public String toString() {
        return String.format(
                "%s[workers:%d][buffer:%d]",
                getClass().getSimpleName(),
                workersRequired(),
                bufferSize()
        );
    }

    protected abstract static class EventEnqueuer<E> extends BaseEnqueuer<E> {

        private final EventFactory<E> eventFactory;


        public EventEnqueuer( final int eventsInBatch,
                              final EventFactory<E> eventFactory ) {
            super( eventsInBatch );
            this.eventFactory = eventFactory;
        }

        @Override
        protected void enqueueEventsBatch( final int eventsInBatch ) throws Exception {
            enqueueEventsBatch( eventsInBatch, eventFactory );
        }

        protected abstract void enqueueEventsBatch( final int eventsInBatch,
                                                    final EventFactory<E> factory ) throws Exception;

    }

}
